package com.mq.kafka;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * @author bobo
 */
@Component
public class ConsumerFunction extends Thread implements ApplicationRunner {

    KafkaConsumer<Integer, String> consumer;
    private static final String topic = "kafka-test";

    public ConsumerFunction() {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer");
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        //自动提交(批量确认)
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //一个新的group的消费者去消费一个topic
        //这个属性. 它能够消费昨天发布的数据
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        consumer = new KafkaConsumer<Integer, String>(properties);
    }

    @Override
    public void run() {
        consumer.subscribe(Collections.singleton(topic));

        while (true) {
            ConsumerRecords<Integer, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
            consumerRecords.forEach(record -> {
                System.out.println(record.key() + "->" + record.value() + "->" + record.offset());
            });
        }
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        System.err.println("================== kafka启动 ==================");
        try {
            Thread thread = new Thread(new ConsumerFunction());
            thread.setDaemon(true);
            thread.start();
        }catch (Exception e){
            System.out.println("kafka抛出异常");
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        new ConsumerFunction().start();
    }
}